package com.zetyun.tiger.sql.kafka;

import cn.hutool.core.io.FileUtil;
import cn.hutool.core.text.StrFormatter;
import cn.hutool.db.Db;
import cn.hutool.db.Entity;
import com.zetyun.tiger.sql.dkm.GenSrDkmApp;

import java.nio.charset.StandardCharsets;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

public class GenFlinkKafkaAllianceDdlApp {

    public static void main(String[] args) throws SQLException {

        ArrayList<String> list = new ArrayList<>();


        String host1 = "36.11.192.111";
        String host2 = "36.11.192.112";
        String host3 = "36.11.192.113";

        //查询
        List<Entity> result = Db.use()
                .query("select concat('ODS_',system_name,'_',tb_name) full_name,tb_name from ods_table ot  " );

        String kSql = FileUtil.readString(Objects.requireNonNull(
                GenSrDkmApp.class.getResource("/primary_key.sql")), StandardCharsets.UTF_8);
        String fieldSql = FileUtil.readString(Objects.requireNonNull(
                GenSrDkmApp.class.getResource("/field.sql")), StandardCharsets.UTF_8);
        String dkmSqlTemplate = FileUtil.readString(Objects.requireNonNull(
                GenSrDkmApp.class.getResource("/flink_kafka.template")), StandardCharsets.UTF_8);

        for (Entity entity : result) {
            String tb_name = entity.getStr("tb_name");
            List<Entity> key = Db.use().query(StrFormatter.format(kSql, tb_name));
            String all_key = key.get(0).getStr("all_key");

            List<Entity> query = Db.use().query(StrFormatter.format(fieldSql,
                    entity.getStr("tb_name"),
                    entity.getStr("tb_name")));

            StringBuilder sb = new StringBuilder();
            StringBuilder sbColumnName = new StringBuilder();
            String tb_cn_name = "";
            for (Entity e : query) {
                tb_cn_name = e.getStr("tb_cn_name");
                String field_def = e.getStr("field_def")
                        .replace("DECIMAL(22)","DECIMAL(22, 0)")
                        .replace("\"","'")
                        ;
                sbColumnName.append(e.getStr("column_name")).append(",");

                sb.append(field_def).append("\n");
            }

            String full_name = entity.getStr("full_name");
            String sql = StrFormatter.format(
                    dkmSqlTemplate,
                    tb_cn_name,
                    full_name,
                    full_name,
                    sb,
                    all_key,
                    tb_cn_name,
                    full_name,
                    host1,
                    host2,
                    host3,
                    full_name,
                    full_name,
                    full_name,
                    sbColumnName,
                    full_name,
                    full_name,
                    full_name,
                    full_name,
                    full_name


            );
            list.add(sql);

//            System.out.println(sql);
//            return;
        }

        FileUtil.writeLines(list,"kafka-ddl-flink.sql",StandardCharsets.UTF_8);
    }
}